package com.lg.rpc.registry.zookeeper;

import com.lg.rpc.common.protocol.util.Constant;
import com.lg.rpc.registry.ServiceRegistry;
import lombok.extern.slf4j.Slf4j;
import org.I0Itec.zkclient.IZkStateListener;
import org.I0Itec.zkclient.ZkClient;
import org.apache.zookeeper.Watcher;

import java.util.concurrent.CountDownLatch;

/**
 * 基于 ZooKeeper 的服务注册接口实现
 *
 * @author by Mr. Li 2020/5/21 23:21
 */
@Slf4j
public class ZooKeeperServiceRegistry implements ServiceRegistry {

    private final CountDownLatch latch = new CountDownLatch(1);


    private final String zkAddress;

    public ZooKeeperServiceRegistry(String zkAddress) {
        // 创建zk客户端
        this.zkAddress = zkAddress;
        log.debug("connect zookeeper...");
    }


    @Override
    public void register(String serviceName, String serviceAddress) {

        String registryPath = Constant.ZK_REGISTRY_PATH + "/" + serviceName + "/address-";
        //// 创建 registry 节点（持久）
        //String registryPath = Constant.ZK_REGISTRY_PATH;
        //if (!zkClient.exists(registryPath)) {
        //    zkClient.createPersistent(registryPath);
        //    log.debug("create registry node: {}", registryPath);
        //}
        //// 创建 service 节点（持久）
        //String servicePath = registryPath + "/" + serviceName;
        //if (!zkClient.exists(servicePath)) {
        //    zkClient.createPersistent(servicePath);
        //    log.debug("create service node: {}", servicePath);
        //}
        //// 创建 address 节点（临时）
        //String addressPath = servicePath + "/address-";
        //String addressNode = zkClient.createEphemeralSequential(addressPath, serviceAddress);

        register(registryPath);

        log.debug("create address node: {}", registryPath);
    }


    /**
     * 升级使用 zkClient
     *
     * @param serverAddress
     */
    @Override
    public void register(String serverAddress) {
        if (serverAddress != null) {
            ZkClient zk = connectServer();
            if (zk != null) {
                AddRootNode(zk); // Add root node if not exist
                createNode(zk, serverAddress);
            }
        }
    }

    /**
     * 升级使用 zkClient
     *
     * @param zk
     * @param serverAddress
     */
    private void createNode(ZkClient zk, String serverAddress) {
        if (zk.exists(Constant.ZK_REGISTRY_PATH))
            zk.createPersistent(Constant.ZK_REGISTRY_PATH);
        // 根据/ 分割路径
        String[] split = serverAddress.split("/");
        StringBuilder spliceStr = new StringBuilder("/");
        for (int i = 0; i < split.length - 1; i++) {
            spliceStr.append(split[i]);
            // 依次创建
            if (i == 2) {
                String servicePath = spliceStr.toString();
                if (zk.exists(servicePath))
                    zk.createPersistent(servicePath);
            }
            if (i != 0)
                spliceStr.append("/");
        }
        // 创建所有路径都为临时路径
        String path = zk.createEphemeralSequential(serverAddress, serverAddress);
        log.debug("create zookeeper node ({} => {})", path, serverAddress);
    }

    /**
     * 升级使用zkClient
     *
     * @param zk
     */
    private void AddRootNode(ZkClient zk) {
        if (zk.exists(Constant.ZK_REGISTRY_PATH)) {
            //zk.create(Constant.ZK_REGISTRY_PATH, new byte[0], ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            zk.createPersistent(Constant.ZK_REGISTRY_PATH);
        }
    }

    /**
     * 连接 zk 客户端
     *
     * @return zk
     */
    private ZkClient connectServer() {
        ZkClient zk = null;
        //ZkClient zk = new ZkClient(new ZkConnection(zkAddress, Constant.ZK_SESSION_TIMEOUT));
        try {
            // 检测 节点的连接状态
            zk = new ZkClient(zkAddress, Constant.ZK_SESSION_TIMEOUT, Constant.ZK_CONNECTION_TIMEOUT);
            zk.subscribeStateChanges(new IZkStateListener() {
                @Override
                public void handleStateChanged(Watcher.Event.KeeperState state) throws Exception {
                    System.out.println("等待");
                    if (state == Watcher.Event.KeeperState.SyncConnected) {
                        latch.countDown();
                    }
                }

                @Override
                public void handleNewSession() throws Exception {

                }

                @Override
                public void handleSessionEstablishmentError(Throwable error) throws Exception {

                }
            });
            latch.await();
        } catch (Exception e) {
            log.error("", e);
        }
        return zk;
    }
}
